package PSQueue;

import java.io.ByteArrayInputStream;
import java.util.ArrayList;
import java.util.HashMap;

import Exception.ExpressionException;
import Expression.Expression;
import Expression.ExpressionValue;
import Service.Variable;

/**
 * @author cyye
 *
 */
public class PSQueue implements AbstractQueue, Runnable {
	
	private ArrayList<AbstractPublication> m_queue = new ArrayList<AbstractPublication>(); //for update of contents
	private synchronized void addPublication(AbstractPublication pub) {
		assert pub!=null;
		m_queue.add(pub);
	}
	private synchronized AbstractPublication getPublication() {
		assert !m_queue.isEmpty();
		return m_queue.remove(0);
	}
	private synchronized boolean hasPublication() {
		return !m_queue.isEmpty();
	}
	
	
	private HashMap<AbstractSubscription, AbstractSubscriber> subs = new HashMap<AbstractSubscription, AbstractSubscriber>(); //for notification
	private synchronized void addSubscription(AbstractSubscription sub, AbstractSubscriber suber) {
		assert sub!=null;
		assert suber!=null;
		subs.put(sub, suber);
	}
	private synchronized AbstractSubscriber getSubscriber(AbstractSubscription sub){
		assert sub!=null;
		return subs.get(sub);
	}
	
	private boolean terminated = false; //for terminating the queue	
	
	private HashMap<String, PSMessage> m_contents = new HashMap<String, PSMessage>(); //for storage of contents
	private synchronized void addContent(String name, PSMessage msg) {
		assert name!=null;
		assert msg!=null;
		m_contents.put(name, msg);
	}
	private synchronized PSMessage getContent(String name) {
		assert name!=null;
		return m_contents.get(name);
	}
	
	
	private HashMap<String, ArrayList<AbstractSubscription>> index = new HashMap<String, ArrayList<AbstractSubscription>>();  //for matching
	private synchronized void addIndex(String id, ArrayList<AbstractSubscription> sublist) {
		assert id!=null;
		assert sublist!=null;
		index.put(id, sublist);
	}
	private synchronized ArrayList<AbstractSubscription> getSubscriptions(String id) { 
		assert id!=null;
		return index.get(id);
	}

	@Override
	public void advertise(AbstractAdvertisement advertisement) {
		// TODO Auto-generated method stub
		assert advertisement != null;
		
	    ArrayList<String> msglist = advertisement.getMsgIDList();
	    assert msglist!=null;
	    for(int i=0;i<msglist.size();i++) {
	    	String msgName = msglist.get(i);	    		    	
	    	PSMessage savedMsg = getContent(msgName);
	    	if(savedMsg == null) {
	    		savedMsg = new PSMessage();
	    		savedMsg.setMsgName(msgName);
	    		
	    		//copy message
	    		AbstractMessage adMsg = advertisement.getMessage(msgName);
	    		assert adMsg!=null;
	    		int type = adMsg.getMsgType();
	    		assert Variable.validType(type);
	    		savedMsg.setMsgType(type);
	    		
	    		Object value = adMsg.getValue();
	    		savedMsg.setValue(value);
	    			    		
	    		addContent(msgName, savedMsg);
	    		
	    	} else {	    		
	    	    System.out.println("Invalid Advertisement: Msg " + msgName + " already exists!");	    		
	    	}
	    }
	}
	
	
	@Override
	public void publish(AbstractPublication pub) {
		// TODO Auto-generated method stub
		assert pub != null;
		
		PSPublication t_pub = new PSPublication();
		ArrayList<String> msgIDs = pub.getMsgIDList();
		assert msgIDs != null;
		
		for(int i=0;i<msgIDs.size(); i++) {
		   String id = msgIDs.get(i);
		   AbstractMessage msg = pub.getMessage(id);
		   assert msg != null;
		   t_pub.addMessage(msg);
		}
		
		addPublication(t_pub);		
	}

	@Override
	public void subscribe(AbstractSubscriber subscriber, AbstractSubscription subscription) {
		// TODO Auto-generated method stub
		assert subscriber != null;
		assert subscription != null;
		
		//copy the subscription
		PSSubscription t_subscription = new PSSubscription();
		ArrayList<String> ids = subscription.getIds();
		assert ids!=null;
		t_subscription.addIDS(ids);
		String condition = subscription.getCondition();
		t_subscription.setCondition(condition);
				
		addSubscription(t_subscription, subscriber);
		
		//create index (msgName->subscription)			
		for(int i=0;i<ids.size();i++) {
			String msgName = ids.get(i);
			ArrayList<AbstractSubscription> msg_index = getSubscriptions(msgName);
            if(msg_index == null) {
            	msg_index = new ArrayList<AbstractSubscription>();
            	addIndex(msgName, msg_index);
            }
            
            synchronized(this) {
                msg_index.add(t_subscription);
            }
		}								
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		while(!this.terminated) {
			if(!hasPublication()) {
				try{
					//System.out.println("Sleep 1 second");
				    Thread.sleep(1000);
				    //System.out.println("Wake up");
				} catch(Exception e) {}								
			} else {
				AbstractPublication pub = getPublication();								
				updateContents(pub);
				//System.out.println("Notify a publication");				
			}
		}		
	}

	private void updateContents(AbstractPublication pub) {
		// TODO Auto-generated method stub
		assert pub!=null;
		
		ArrayList<AbstractSubscription> need_check = new ArrayList<AbstractSubscription>();			
		ArrayList<String> msgIDlist = pub.getMsgIDList();
		
		//check whether it is valid
		Boolean valid = true;
		for(int i=0;i<msgIDlist.size();i++) {
			String msgName  = msgIDlist.get(i);
			AbstractMessage msg = pub.getMessage(msgName);
			assert msg!=null;
			
			int type = msg.getMsgType();
			assert Variable.validType(type);
			
			AbstractMessage savedMsg = getContent(msgName);			
			if((savedMsg != null) && (savedMsg.getMsgType() != type)) {
				valid = false;
				break;
			}
			
		}
		
		if(valid) {			
			//update contents
			for(int i=0;i<msgIDlist.size();i++) {
				String msgName  = msgIDlist.get(i);
				AbstractMessage msg = pub.getMessage(msgName);							
				Object value = msg.getValue();						
			    assert value!=null;
			    
			    PSMessage savedMsg = getContent(msgName);							    			
				if(savedMsg != null) {
					//update contents
					savedMsg.setValue(value);
					ArrayList<AbstractSubscription> index_subs = getSubscriptions(msgName);
					if(index_subs!=null) {
					   need_check.removeAll(index_subs);
					   need_check.addAll(index_subs);
					}
				} 							
			}
			
			//matching
			matchSubscriptions(need_check);
		}
	}

	private void matchSubscriptions(ArrayList<AbstractSubscription> needCheck) {
		// TODO Auto-generated method stub
		assert needCheck!=null;
		AbstractPublication notification;
		for(int i=0;i<needCheck.size();i++) {
			AbstractSubscription subscription = needCheck.get(i);			
			notification = matching(subscription);
			if(notification!=null) {
				AbstractSubscriber subscriber = getSubscriber(subscription);				
				assert subscriber!=null;
				sendNotification(subscriber, notification);
			}
			
		}
	}

	private void sendNotification(AbstractSubscriber subscriber, AbstractPublication notification) {
		// TODO Auto-generated method stub
		//use multiple threads
		assert subscriber!=null;
		assert notification!=null;
		new Notifier(subscriber, notification).start();
	}

	private AbstractPublication matching(AbstractSubscription subscription) {
		// TODO Auto-generated method stub
		assert subscription != null;
		
		PSPublication notification = new PSPublication();
		
		ArrayList<String> ids = subscription.getIds();
		assert ids!=null;
		
		ArrayList<Variable> variables = new ArrayList<Variable>();
		Variable variable;
		PSMessage msg;
		for(int i=0;i<ids.size();i++) {
			String msgName = ids.get(i);
			PSMessage savedMsg = getContent(msgName);
			if(savedMsg ==null) return null;//no advertisement of this message
			
			int type = savedMsg.getMsgType();
			assert Variable.validType(type);
			
			Object value = savedMsg.getValue();
			if(value == null) {
				return null; //not ready for matching				
			}
			
			variable = new Variable();
			variable.setName(msgName);
			variable.setType(type);
			variable.setValue(value);
			variables.add(variable);
			
			//note that the message should only be readable for subscribers
			msg = new PSMessage();
			msg.setMsgName(msgName);
			msg.setMsgType(type);
			msg.setValue(value);
			notification.addMessage(msg);
		}
		
		//all the contents are ready
		//matching the condition
		String condition = subscription.getCondition();
		assert condition!=null;
		
		ByteArrayInputStream in = new ByteArrayInputStream(condition.getBytes());	   
		Expression parser = new Expression(in);
		ExpressionValue result;
		   
		try{
			result = parser.CalculateExpression(variables);		
			if((result==null) || (result.type != Variable.BOOLEAN)) throw new ExpressionException("Invalid Subscription condition!");
		    if(((Boolean)result.value).booleanValue())
		        return notification; //matched
		    else
		    	return null; //not matched
			   
		} catch (Exception e) {
		   System.out.println("NOK.");
		   System.out.println(e);				        
		} catch (Error e) {
		   System.out.println("Oops.");
		   System.out.println(e);		
		}
		
		return null;
	}

	public void turnOffQueue() {
		this.terminated  = true;
	}
	
	
	/*
	 * --------------------------------------------------------------------------------------
	 *  The following code is for unit testing
	 * --------------------------------------------------------------------------------------
	 */
	
	public void testPSPublication() {
		String msgName1 = "msg1";
		String msgValue1 = "testcase1";
		
		String msgName2 = "msg2";
		String msgValue2 = "testcase2";
		
		PSPublication pub = new PSPublication();
		PSMessage msg = new PSMessage();
		
		msg.setMsgName(msgName1);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue1);
		pub.addMessage(msg);		
		
		msg.setMsgName(msgName2);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue2);
		pub.addMessage(msg);
		
		msg.setMsgName(msgName2);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue1);
		pub.addMessage(msg);
		
		ArrayList<String> ids = pub.getMsgIDList();
		assert(ids.size() == 2);
		
		AbstractMessage t_msg = pub.getMessage(msgName1);
		assert(t_msg!=null);
		assert(msgValue1.equals(t_msg.getValue()));
		
		t_msg = pub.getMessage(msgName2);
		assert(t_msg!=null);
		assert(msgValue2.equals(t_msg.getValue()));				
	}
		
	public void testPSSubscription() {
		String msgName1 = "msg1";				
		String msgName2 = "msg2";		
		String msgName3 = "msg3";
		String msgName4 = "msg4";
		String msgName5 = "msg5";
		
		PSSubscription sub = new PSSubscription();
		sub.addID(msgName1);
		sub.addID(msgName2);
		
		ArrayList<String> idlist = new ArrayList<String>();
		idlist.add(msgName2);
		idlist.add(msgName3);
		idlist.add(msgName4);
		idlist.add(msgName5);
		
		sub.addIDS(idlist);
		sub.addID(msgName4);
		
		ArrayList<String> subIDs = sub.getIds();
		assert(subIDs.size() == 5);
		
		assert(subIDs.contains(msgName1));
		assert(subIDs.contains(msgName2));
		assert(subIDs.contains(msgName3));
		assert(subIDs.contains(msgName4));
		assert(subIDs.contains(msgName5));
	}
	
	public void testPSAdvertisement() {
		String msgName1 = "msg1";
		String msgValue1 = "testcase1";
		
		String msgName2 = "msg2";
		String msgValue2 = "testcase2";
		
		PSAdvertisement adv = new PSAdvertisement();
		PSMessage msg = new PSMessage();
		
		msg.setMsgName(msgName1);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue1);
		adv.addAdvertisement(msg);		
		
		msg.setMsgName(msgName2);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue2);
		adv.addAdvertisement(msg);
		
		msg.setMsgName(msgName1);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue2);
		adv.addAdvertisement(msg);
		
		ArrayList<String> ids = adv.getMsgIDList();
		assert(ids.size() == 2);
		
		AbstractMessage t_msg = adv.getMessage(msgName1);
		assert(t_msg!=null);
		assert(msgValue1.equals(t_msg.getValue()));
		
		t_msg = adv.getMessage(msgName2);
		assert(t_msg!=null);
		assert(msgValue2.equals(t_msg.getValue()));					
	}
	
	
	public void testAddAdvertisement() {
		String msgName1 = "msg1";
		String msgValue1 = "testcase1";
		
		String msgName2 = "msg2";
		String msgValue2 = "testcase2";
		
		String msgName3 = "msg3";
		String msgValue3 = "testcase3";
		
		PSAdvertisement adv = new PSAdvertisement();
		PSMessage msg = new PSMessage();
		
		msg.setMsgName(msgName1);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue1);
		adv.addAdvertisement(msg);		
		
		msg.setMsgName(msgName2);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue2);
		adv.addAdvertisement(msg);				
		advertise(adv);
		
		adv = new PSAdvertisement(); 
		msg.setMsgName(msgName1);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue2);
		adv.addAdvertisement(msg);
		
		msg.setMsgName(msgName3);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue3);
		adv.addAdvertisement(msg);
		advertise(adv);
		
		assert(getContent(msgName1)!=null);
		assert(getContent(msgName2)!=null);
		assert(getContent(msgName3)!=null);		
	}
	
	public void testAddSubscription() {
		String msgName1 = "msg1";				
		String msgName2 = "msg2";		
		String msgName3 = "msg3";
		String msgName4 = "msg4";
		String msgName5 = "msg5";
		
		PSSubscription sub1 = new PSSubscription();
		sub1.addID(msgName1);
		sub1.addID(msgName2);		
		
		SampleSubscriber subscriber1 = new SampleSubscriber(); 
		subscribe(subscriber1, sub1);
		
		PSSubscription sub2 = new PSSubscription();
		sub2.addID(msgName3);
		
		SampleSubscriber subscriber2 = new SampleSubscriber(); 
		subscribe(subscriber2, sub2);
		
		ArrayList<String> idlist = new ArrayList<String>();
		idlist.add(msgName2);
		idlist.add(msgName3);
		idlist.add(msgName4);
		idlist.add(msgName5);		
		
		PSSubscription sub3 = new PSSubscription();
		sub3.addIDS(idlist);
		sub3.addID(msgName4);
		
		SampleSubscriber subscriber3 = new SampleSubscriber(); 
		subscribe(subscriber3, sub3);
		
		//test
		ArrayList<AbstractSubscription> subs = getSubscriptions(msgName1);
		assert(subs.size()==1);
		//assertTrue(subs.contains(sub1));
		assert (subscriber1.equals(getSubscriber(subs.get(0))));
		
		subs = getSubscriptions(msgName2);
		assert (subs.size()==2);
		//assertTrue(subs.contains(sub1));
		//assertTrue(subs.contains(sub3));
		
		subs = getSubscriptions(msgName3);
		assert(subs.size()==2);
		assert(subscriber2.equals(getSubscriber(subs.get(0)))||subscriber2.equals(getSubscriber(subs.get(1))));
		//assertTrue(subs.contains(sub2));
		//assertTrue(subs.contains(sub3));
		
		subs = getSubscriptions(msgName4);
		assert(subs.size()==1);		
		//assertTrue(subs.contains(sub3));
		
		subs = getSubscriptions(msgName5);
		assert(subs.size()==1);		
		assert(subscriber3.equals(getSubscriber(subs.get(0))));
		//assertTrue(subs.contains(sub3));						
		
	}
	
    public void testAddPublication() {
    	String msgName1 = "msg1";	
    	String msgValue1 = "value1";
		String msgName2 = "msg2";	
		String msgValue2 = "value2";
		String msgName3 = "msg3";
		String msgValue3 = "value3";
		String msgName4 = "msg4";
		String msgValue4 = "value4";
		String msgName5 = "msg5";
		String msgValue5 = "value5";
		
		PSPublication pub = new PSPublication();
		PSMessage msg = new PSMessage();
		
		msg.setMsgName(msgName1);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue1);
		pub.addMessage(msg);		
		
		msg.setMsgName(msgName2);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue2);
		pub.addMessage(msg);
		
		publish(pub);
		
		msg.setMsgName(msgName3);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue3);
		pub.addMessage(msg);
		publish(pub);
		
		pub = new PSPublication();
		msg.setMsgName(msgName3);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue3);
		pub.addMessage(msg);
		
		msg.setMsgName(msgName4);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue4);
		pub.addMessage(msg);
		
		msg.setMsgName(msgName5);
		msg.setMsgType(Variable.STRING);
		msg.setValue(msgValue5);		
		pub.addMessage(msg);		
		publish(pub);
		
		//test
		//publication 1
		AbstractPublication t_pub = getPublication();
		AbstractMessage t_msg = t_pub.getMessage(msgName1);
		assert t_msg!=null;
		assert msgValue1.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName2);
		assert t_msg!=null;
		assert msgValue2.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName3);
		assert t_msg==null;
		
		//publication 2
		t_pub = getPublication();
		t_msg = t_pub.getMessage(msgName1);
		assert t_msg!=null;	
		assert msgValue1.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName2);
		assert t_msg!=null;
		assert msgValue2.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName3);
		assert t_msg!=null;
		assert msgValue3.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName4);
		assert t_msg==null;
		
		//publication3
		t_pub = getPublication();
		t_msg = t_pub.getMessage(msgName1);
		assert t_msg==null;		
		
		t_msg = t_pub.getMessage(msgName3);
		assert t_msg!=null;
		assert msgValue3.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName4);
		assert t_msg!=null;
		assert msgValue4.equals(t_msg.getValue());
		
		t_msg = t_pub.getMessage(msgName5);
		assert t_msg!=null;
		assert msgValue5.equals(t_msg.getValue());		
		
		assert !hasPublication();		
	}	
	
}

class Notifier extends Thread {
	
	AbstractSubscriber m_subscriber;
	AbstractPublication m_notification;
	public Notifier(AbstractSubscriber subscriber, AbstractPublication notification) {
		assert subscriber!=null;
		assert notification!=null;
		m_subscriber = subscriber;
		m_notification = notification;
	}
  
	public void run() {
		m_subscriber.OnNotification(m_notification);
	}
}